package com.ada.spark.accumulator

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}
import scala.collection.JavaConversions._

/**
  * 自定义累加器类型的功能在1.X版本中就已经提供了，
  * 但是使用起来比较麻烦，在2.0版本后，累加器的易用性有了较大的改进，
  * 而且官方还提供了一个新的抽象类：AccumulatorV2来提供更加友好的自定义类型累加器的实现方式。
  * 实现自定义类型累加器需要继承AccumulatorV2并至少覆写下例中出现的方法，
  * 下面这个累加器可以用于在程序运行过程中收集一些文本类信息，最终以Set[String]的形式返回。
  */
class LogAccumulator extends org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] {
    private val _logArray: java.util.Set[String] = new java.util.HashSet[String]()

    override def isZero: Boolean = {
        _logArray.isEmpty
    }

    override def reset(): Unit = {
        _logArray.clear()
    }

    override def add(v: String): Unit = {
        _logArray.add(v)
    }

    override def merge(other: org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]]): Unit = {
        other match {
            case o: LogAccumulator => _logArray.addAll(o.value)
        }

    }

    override def value: java.util.Set[String] = {
        java.util.Collections.unmodifiableSet(_logArray)
    }

    override def copy(): org.apache.spark.util.AccumulatorV2[String, java.util.Set[String]] = {
        val newAcc = new LogAccumulator()
        _logArray.synchronized {
            newAcc._logArray.addAll(_logArray)
        }
        newAcc
    }
}

// 过滤掉带字母的
object LogAccumulator {
    def main(args: Array[String]) {
        //获取Spark配置信息并创建与spark的连接
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("LogAccumulator")
        val sc = new SparkContext(sparkConf)

        val accum = new LogAccumulator
        sc.register(accum, "logAccum")
        val sum = sc.parallelize(Array("1", "2a", "3", "4b", "5", "6", "7cd", "8", "9"), 2).filter(line => {
            val pattern = """^-?(\d+)"""
            val flag = line.matches(pattern)
            if (!flag) {
                accum.add(line)
            }
            flag
        }).map(_.toInt).reduce(_ + _)

        println("sum: " + sum) //sum: 32
        for (v <- accum.value) print(v + "") //7cd4b2a

        println()
        sc.stop()
    }
}
